package web

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.Directives._
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Sink
import akka.stream.scaladsl.{Flow, Source}
import akka.stream.actor._
import akka.actor._
import akka.http.scaladsl.model.HttpEntity.Strict
import akka.http.javadsl.model.ws._
import akka.stream.scaladsl.Sink
import scala.concurrent.forkjoin.ThreadLocalRandom

import akka.http.scaladsl.model.ws.{TextMessage, Message, BinaryMessage}

import abx.protos.peer._
import akka.util.ByteString


case class Tick()

//class TickActor extends ActorPublisher[Int] {
class TickActor extends ActorPublisher[Book] {
  import scala.concurrent.duration._

  implicit val ec = context.dispatcher

  val tick = context.system.scheduler.schedule(2 second, 2 second, self, Tick())

  var cnt = 0
  //var buffer = Vector.empty[Int]
  var buffer = Vector.empty[Book]

  override def receive: Receive = {
    case Tick() => {
      cnt = cnt + 1
      val b = Book("book "+cnt)
     // b.toByteArray
     // b.toByteString
      if (buffer.isEmpty && totalDemand > 0) {
        //onNext(cnt)
        onNext(b)
      }
      else {
        //buffer :+= cnt
        buffer :+= b
        if (totalDemand > 0) {
          val (use,keep) = buffer.splitAt(totalDemand.toInt)
          buffer = keep
          use foreach onNext
        }
      }
    }
  }

  override def postStop() = tick.cancel()
}


object WebServer {

  implicit val system = ActorSystem("akka-social-stream-example")
  implicit val materializer = ActorMaterializer()

  def main(args: Array[String]) {
    val bm = BinaryMessage(ByteString("abcdef"))
    val route =
      pathEndOrSingleSlash {
        get {
          getFromResource("pw1.html")
        }
      } ~
     (get & pathPrefix("web")) {
        getFromResourceDirectory("web")
      }~
      path("feed") {
        get {
         //val source = Source.actorPublisher[Book](Props[TickActor]).map(i => TextMessage(i.toString)) 
         // val source = Source.actorPublisher[Book](Props[TickActor]).map(i =>  BinaryMessage(ByteString("abcdef")) )
         val source = Source.actorPublisher[Book](Props[TickActor]).map(i =>  BinaryMessage(ByteString(i.toByteArray)))   
          extractUpgradeToWebSocket { upgrade =>
            complete(upgrade.handleMessagesWithSinkSource(Sink.ignore, source))
          }
        }
      }

    Http().bindAndHandle(route, "localhost", 8080)
    println(s"Server online at http://localhost:8080/")
  }

}
